package com.heima.article.listen;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.heima.article.mapper.ApArticleConfigMapper;
import com.heima.article.mapper.ApArticleMapper;
import com.heima.model.article.entity.ApArticle;
import com.heima.model.article.entity.ApArticleConfig;
import com.heima.model.constants.KafkaMessageConstants;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Map;

@Component
@Slf4j
public class KafkaArticleListener {

    @Resource
    private ApArticleMapper apArticleMapper;

    @Resource
    private ApArticleConfigMapper apArticleConfigMapper;


    /**
     * 文章上下架处理
     * @param record
     */
    @KafkaListener(topics = KafkaMessageConstants.WM_NEWS_UP_OR_DOWN_TOPIC,groupId ="WM_NEWS" )
    public void upOrDownArticle(ConsumerRecord<String,String> record){
        String value = record.value();
        log.info("接收到的消息为：{}",value);
        if(StringUtils.isEmpty(value)){
            return;
        }

        //解析消息
        Map<String,String> map = JSON.parseObject(value, Map.class);
        Long newsId = Long.valueOf(map.get("newsId"));
        Short enable = Short.valueOf(map.get("enable"));
        //根据newsId 查询文章
        ApArticle apArticle = apArticleMapper.selectOne(Wrappers.<ApArticle>lambdaQuery().eq(ApArticle::getNewsId, newsId));
        if(apArticle==null){
            return;
        }
        //查询文章配置信息
        ApArticleConfig articleConfig = apArticleConfigMapper.selectOne(Wrappers.<ApArticleConfig>lambdaQuery().eq(ApArticleConfig::getArticleId, apArticle.getId()));
        if (articleConfig == null) {
            return;
        }
        //更新文章配置信息
        articleConfig.setEnable(enable.intValue());
        apArticleConfigMapper.updateById(articleConfig);

    }
}
